package com.zhss.seckill.common;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;

public class RocketMQConsumer {

    private RocketMQConsumer(String consumerGroup, String topic,
                             MessageListenerConcurrently listener) {
        try {
            System.out.println(consumerGroup);
            System.out.println(topic);
            System.out.println(listener);

            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr("localhost:9876");
            consumer.subscribe(topic, "*");
            consumer.registerMessageListener(listener);
            consumer.start();
            System.out.println("RocketMQ消费者启动成功......");
        } catch(MQClientException e) {
            System.err.println("初始化RocketMQ消费者失败......" + e);
        }
    }

    private static class Singleton {

        static RocketMQConsumer instance = new RocketMQConsumer(
                consumerGroup,
                topic,
                listener
        );

    }

    public RocketMQConsumer getInstance() {
        return Singleton.instance;
    }

    private static String consumerGroup;
    private static String topic;
    private static MessageListenerConcurrently listener;

    public static void setConsumerGroup(String consumerGroup) {
        RocketMQConsumer.consumerGroup = consumerGroup;
    }

    public static void setTopic(String topic) {
        RocketMQConsumer.topic = topic;
    }

    public static void setListener(MessageListenerConcurrently listener) {
        RocketMQConsumer.listener = listener;
    }

    public static RocketMQConsumer init() {
        return Singleton.instance;
    }

}
